import asyncio
import hashlib
import hmac
import json
import logging
import threading
import time
from dataclasses import dataclass
from datetime import datetime
from typing import List, Optional, Dict, Iterable

import cacheout
import pytz
import requests

from common.utils.tools import str2float
from libs.deribit import configuration, api_client, PrivateApi, ApiException, rest
from libs.deribit.api import public_api
from .base import ExchangePublicServiceBase, ExchangePriServiceBase, ExchangeErrorCode, \
    InstrumentQuote, ExchangeQuotesCatcher, ExchangeWSRunner, generate_found_rate_by_quote, ExchangeUserDataCatcher, \
    AsyncRequest
from .contants import Exchange, ContratType, OptionType, OrderType, AuthInfoType, STABLE_COINS, OrderState, FutureType
from .entities import SimpleInstrument, Instrument, OrderBook, Ticker, IndexPair, \
    IndexPrice, OrderBookItem, Trade, Order, Position, CurrencySummary, DeribitOauthAuthInfo, ApiAuthInfo, \
    Option, Future, Spot, AuthInfo
from ..entities import FutureFundRateDetail

EXCHANGE_CONFIG = {
    'test': {
        # 'api_host': 'https://dev.yazhan.vip/transfer/api/v2',
        'api_host': 'https://test.deribit.com/api/v2',
        'ws_host': 'wss://test.deribit.com/ws/api/v2/'
    },
    'greeks_test': {
        'api_host': 'https://test.greeks.live/transfer/api/v2',
        'ws_host': 'wss://test.greeks.live/transfer/ws/api/v2/'
    },
    'prod': {
        'api_host': 'https://asia.deribit.com/api/v2',
        'ws_host': 'wss://asia.deribit.com/ws/api/v2/'
    },
    'greeks_prod': {
        'api_host': 'https://www.greeks.live/transfer/api/v2',
        'ws_host': 'wss://www.greeks.live/transfer/ws/api/v2/'
    }
}


class DeribitWSRunner(ExchangeWSRunner):

    async def _handle_subscription_msg(self, channel, data):
        raise NotImplementedError

    async def _on_connected(self):
        raise NotImplementedError

    @classmethod
    def _parse_resp_msg(cls, message: Dict):
        if 'error' in message:
            error = message['error']
            return ApiException(status=error['code'], reason=error['message'])
        return message.get('result')

    async def _on_message(self, message):
        method = message.get('method', '')
        msg_id = message.get('id')

        # request response 类型消息
        if msg_id and msg_id in self._requests:
            request: AsyncRequest = self._requests.pop(msg_id)
            if request:
                try:
                    await request.set_result(self._parse_resp_msg(message))
                except Exception as e:
                    self._logger.exception(f'{self.name} handle callback for {message} except: {e}')

        # 订阅类消息
        elif method == 'subscription':
            params = message.get('params', {})
            channel: str = params.get('channel', '')
            data = params.get('data')
            await self._handle_subscription_msg(channel, data or {})
        else:
            self._logger.warning(f'{self.name} got can not handle message: {message}')

    async def _subscribe(self, channels, is_pub=True):
        start, step = 0, 100
        method = 'public/subscribe' if is_pub else 'private/subscribe'
        while start * step < len(channels):
            await self._request(
                method, {'channels': channels[start*step:(start+1)*step]}, callback=self._on_subscribe)
            start += 1

    def _on_subscribe(self, result, ):
        if isinstance(result, Iterable):
            self._subscribed_set.update(set(result))
        self._logger.info(f'{self.name} current subscription cnt is {len(self._subscribed_set)}')

    async def _unsubscribe(self, channels, is_pub=False):
        start, step = 0, 100
        method = 'public/unsubscribe' if is_pub else 'private/unsubscribe'
        while start * step < len(channels):
            await self._request(
                method, {'channels': channels[start*step:(start+1)*step]}, callback=self._on_unsubscribe)
            start += 1

    def _on_unsubscribe(self, result, ):
        if isinstance(result, Iterable):
            self._subscribed_set -= set(result)
        self._logger.info(f'cancel channels: {result}')

    async def _request(self, method, params, callback=None) -> AsyncRequest:
        req_id = self._request_id
        payloads = {
            "jsonrpc": "2.0",
            "id": req_id,
            "method": method,
            "params": params
        }
        if not callable(callback):
            self._logger.warning(f'{self.platform} catcher request {method}, params {params}, callback is not callable')
            callback = None

        await self._client.send(json.dumps(payloads))

        request = AsyncRequest(
            id=req_id, payloads=payloads, callback=callback
        )
        self._requests[req_id] = request
        return request


class DbtQuotesCatcher(ExchangeQuotesCatcher, DeribitWSRunner):

    platform = Exchange.DERIBIT

    def __init__(self, env, currencies=None, margin_currencies=None, logger=None, *args, **kwargs):

        if env not in EXCHANGE_CONFIG:
            raise ValueError(f'{self.platform} got invalid args for env: {env}')
        ws_url = EXCHANGE_CONFIG[env]['ws_host']

        DeribitWSRunner.__init__(
            self, env=env, ws_url=ws_url, logger=logger,
            *args, **kwargs
        )
        ExchangeQuotesCatcher.__init__(self, *args, **kwargs)

        self.currencies = currencies or ['BTC', 'ETH']
        self._margin_currencies = margin_currencies or ['BTC', 'ETH']

    @property
    def name(self):
        return 'eqc:dbt'

    async def _on_connected(self):
        # 生成待订阅频道
        channels = []
        for currency in self._margin_currencies:
            channels.append(f'instrument.state.any.{currency}')

            # 获取合约列表，并订阅数据
            await self._request('public/get_instruments', {
                'currency': currency, 'kind': 'any',
            }, callback=self._on_get_instruments)
        await self._subscribe(channels)

    async def _on_disconnect(self):
        await super()._on_disconnect()
        # 断连时仅清理行情数据，不清理合约信息

        for quote in self._quotes.values():
            quote.clear()
        self._index_prices.clear()

    async def _on_get_instruments(self, result):
        channels = set()
        result = result if isinstance(result, list) else [result]

        for instrument in result:
            if instrument.get('base_currency') not in self.currencies:
                continue
            if instrument.get('kind') not in {'future', 'option', 'spot'}:
                continue

            inst_info: Instrument = DeribitTools.transfer_instrument_data(instrument)
            if not inst_info:
                continue

            quote = self._quotes.setdefault(instrument['instrument_name'], InstrumentQuote(inst_info=inst_info))
            quote.inst_info = inst_info

            # 订阅指数价格
            price_index = instrument.get('price_index')
            price_index_channel = f'deribit_price_index.{price_index}'
            if price_index and price_index_channel not in self._subscribed_set:
                channels.add(price_index_channel)

            # 订阅期货行情数据
            if inst_info.kind == ContratType.Future:
                channels.add(f'ticker.{instrument["instrument_name"]}.100ms')

            if instrument.get('settlement_period') == 'perpetual':
                per_interest_channel = f'perpetual.{instrument["instrument_name"]}.{100}ms'
                if per_interest_channel not in self._subscribed_set:
                    channels.add(per_interest_channel)

        if channels:
            await self._subscribe(list(channels))

    async def _handle_subscription_msg(self, channel: str, data):
        try:
            if not channel:
                self._logger.error(f'can not handle subscription without channel info, data is {data}')
                return

            channel_info = channel.split('.')

            if channel_info[0] == 'ticker' and len(channel_info) == 3:
                instrument = channel_info[1]
                if instrument in self._quotes:
                    self._quotes[instrument].mark_price = data['mark_price']
                    self._quotes[instrument].index_price = data['index_price']

            elif channel_info[0] == 'deribit_price_index' and len(channel_info) == 2:
                index_name = channel_info[1]
                self._index_prices[index_name] = IndexPrice(price=data['price'], timestamp=data['timestamp'])
            elif channel_info[0] == 'perpetual' and len(channel_info) == 3:
                instrument = channel_info[1]
                if instrument in self._quotes:
                    self._quotes[instrument].interest_rate = data['interest']
            elif '.'.join(channel_info[:2]) == 'instrument.state' and len(channel_info) == 4:
                if data.get('state') == 'created':
                    inst_info = DeribitPublicService.parse_inst(data['instrument_name'])
                    if not inst_info:
                        return
                    if inst_info.margin_currency not in self._margin_currencies or inst_info.base_currency not in self.currencies:
                        return

                    await self._request(
                        'public/get_instrument',
                        {'instrument_name': data["instrument_name"]},
                        callback=self._on_get_instruments
                    )
                elif data.get('state') in ['closed']:
                    self._quotes.pop(data["instrument_name"], None)
                    ticker_channel = f'ticker.{data["instrument_name"]}.100ms'
                    if ticker_channel in self._subscribed_set:
                        await self._unsubscribe([ticker_channel])
        except Exception as e:
            self._logger.exception(f'{self.name} handle subscription {channel} msg {data} except: {e}')

    def get_instrument(self, instrument, **kwargs):
        quote = self._quotes.get(instrument)
        return quote.inst_info if quote else None

    def get_index_price(self, pair, active_ts=1000):
        now_ts = int(time.time() * 1000)
        if not self.is_available() or (now_ts - self._last_actively_ts) > active_ts:
            return None
        index_price = self._index_prices.get(pair)
        return index_price

    def mark_price(self, instrument, **kwargs):
        now_ts = int(time.time() * 1000)
        if not self.is_available() or (now_ts - self._last_actively_ts) > 3000:
            return None
        res = self._quotes.get(instrument)
        return res.mark_price if res else None

    def get_future_found_rate_details(self, ) -> List[FutureFundRateDetail]:
        if not self.is_available():
            return []

        results = []
        for instrument, quote in self._quotes.items():
            found_rate = generate_found_rate_by_quote(quote, self.platform)
            if found_rate:
                results.append(found_rate)
        return results


class DeribitTools:
    EXCHANGE_CODE_TO_GREEKS_CODE_MAP = {
        'default': ExchangeErrorCode.ServerUnavailable,

        10019: ExchangeErrorCode.ServerUnavailable,  # Trading temporarily locked by admin
        10028: ExchangeErrorCode.ServerRateLimit,  # Too many requests
        11051: ExchangeErrorCode.ServerMaintenance,  # System under maintenance
        13888: ExchangeErrorCode.OperationTimeout,  # Request timeout
        11094: ExchangeErrorCode.ServerInternalError,  # Internal server error

        # 认证相关错误
        10000: ExchangeErrorCode.AuthFailed,  # Authorization required
        11042: ExchangeErrorCode.PermissionDenied,  # Permission denied
        13004: ExchangeErrorCode.InvalidCredentials,  # Invalid credentials
        13009: ExchangeErrorCode.TokenExpired,  # Expired token

        # 请求参数相关错误
        10001: ExchangeErrorCode.InvalidParameter,  # General failure
        10020: ExchangeErrorCode.InvalidSymbol,  # Invalid instrument
        10023: ExchangeErrorCode.InvalidPrice,  # Invalid price
        10005: ExchangeErrorCode.ParameterOutOfRange,  # Price too low
        10007: ExchangeErrorCode.ParameterOutOfRange,  # Price too high
        10022: ExchangeErrorCode.InvalidQuantity,  # Invalid quantity
        10024: ExchangeErrorCode.InvalidFormat,  # Invalid max_show parameter
        11050: ExchangeErrorCode.InvalidParameter,  # Bad request
        13020: ExchangeErrorCode.InvalidSymbol,

        # 交易相关错误
        10002: ExchangeErrorCode.InvalidQuantity,  # Order quantity too low
        10004: ExchangeErrorCode.OrderNotExist,  # Order not found
        10009: ExchangeErrorCode.MarginInsufficient,  # Insufficient funds
        10011: ExchangeErrorCode.OrderRejected,  # Price not allowed
        13010: ExchangeErrorCode.MissingParameter,  # Missing required value (修正)
        10003: ExchangeErrorCode.SelfTradingRejected,  # Overlap in orders (self-trading not allowed)
        10010: ExchangeErrorCode.ActionOnClosedOrder,  # Action attempted on a closed order
        11054: ExchangeErrorCode.PostOnlyOrderRejected,

        # 账户相关错误
        10012: ExchangeErrorCode.AccountDisabled,  # Order book closed
        11098: ExchangeErrorCode.AccountLocked,  # Account locked
        11095: ExchangeErrorCode.WithdrawDisabled,  # Withdrawals disabled
        11097: ExchangeErrorCode.NoDepositAddress,  # Missing deposit address
        12001: ExchangeErrorCode.AccountInLiquidation,  # Too many subaccounts

        # 网络与其他错误
        10040: ExchangeErrorCode.Retry,  # Request should be retried
        10048: ExchangeErrorCode.ServerUnavailable,  # Operation not available on this server
        13028: ExchangeErrorCode.OperationTimeout,  # Service temporarily unavailable
    }

    @classmethod
    def transfer_error(cls, error):
        if not isinstance(error, ApiException):
            return error
        error_body = json.loads(error.body)
        if error_body.get('code') not in cls.EXCHANGE_CODE_TO_GREEKS_CODE_MAP:
            error_code = cls.EXCHANGE_CODE_TO_GREEKS_CODE_MAP['default']
        else:
            error_code = cls.EXCHANGE_CODE_TO_GREEKS_CODE_MAP[error_body.get('code')]
        return error_code.to_error()

    @classmethod
    def transfer_ticker_data(cls, ticker_data: Dict):
        result = Ticker(
            mark_price=ticker_data.get('mark_price'), timestamp=ticker_data.get('timestamp'),
            index_price=ticker_data.get('index_price'), best_bid_price=ticker_data.get('best_bid_price'),
            best_bid_size=ticker_data.get('best_bid_amount'), best_ask_size=ticker_data.get('best_ask_amount'),
            best_ask_price=ticker_data.get('best_ask_price')
        )
        if ticker_data.get('greeks'):
            greeks = ticker_data['greeks']
            result.delta = greeks['delta']
            result.gamma = greeks['gamma']
            result.theta = greeks['theta']
            result.vega = greeks['vega']

        return result

    @classmethod
    def transfer_instrument_kind(cls, kind):
        if kind == 'option':
            return ContratType.Option
        if kind == 'spot':
            return ContratType.Spot
        if kind == 'future':
            return ContratType.Future
        return kind

    @classmethod
    def transfer_instrument_data(cls, instrument_data) -> Optional[Instrument]:
        price_index = instrument_data.get('price_index', '')
        index_info = price_index.split('_') if price_index else []
        index_pair = None
        if index_info and len(index_info) == 2:
            index_pair = IndexPair(base=index_info[0].upper(), quote=index_info[1].upper())

        # 排除掉combo类型合约信息
        if instrument_data['kind'] in ['future_combo', 'option_combo']:
            return

        instrument_cls = Instrument
        kind = cls.transfer_instrument_kind(instrument_data['kind'])
        if kind == ContratType.Option:
            instrument_cls = Option
        if kind == ContratType.Future:
            instrument_cls = Future
        if kind == ContratType.Spot:
            instrument_cls = Spot

        item = instrument_cls(
            instrument_id=instrument_data['instrument_id'],
            instrument_name=instrument_data['instrument_name'],
            exchange=Exchange.DERIBIT, base_currency=instrument_data['base_currency'],
            quote_currency=instrument_data['quote_currency'], counter_currency=instrument_data['counter_currency'],
            margin_currency=instrument_data.get('settlement_currency') or instrument_data['quote_currency'],
            min_trade_size=instrument_data['min_trade_amount'], trade_size_step=instrument_data['min_trade_amount'],
            min_tick_size=instrument_data['tick_size'], tick_size_step=instrument_data['tick_size'],
            index_pair=index_pair, contract_multi=1,
            kind=kind
        )
        if isinstance(item, Option):
            item.strike = instrument_data['strike']
            item.reversed = instrument_data.get('instrument_type') == 'reversed'
            item.expire_timestamp = instrument_data['expiration_timestamp']
            item.option_type = instrument_data['option_type']
        if isinstance(item, Future):
            item.reversed = instrument_data.get('instrument_type') == 'reversed'
            if instrument_data['settlement_period'] == 'perpetual':
                item.future_type = FutureType.Perpetual
            else:
                item.future_type = FutureType.Delivery
                item.expire_timestamp = instrument_data['expiration_timestamp']
        return item

    @classmethod
    def parse_inst(cls, inst_name) -> Optional[SimpleInstrument]:
        months = ["JAN", "FEB", "MAR", "APR", "MAY", "JUN", "JUL", "AUG", "SEP", "OCT", "NOV", "DEC"]

        try:
            res = SimpleInstrument(instrument=inst_name)

            parts = inst_name.split("-")
            curr_info = parts[0].split('_')
            datestr = None

            res.base_currency = parts[0]
            res.margin_currency = curr_info[0]
            res.reversed = True
            if len(curr_info) > 1 and curr_info[-1] in ['USDC', 'USDT']:
                res.reversed = False
                res.margin_currency = curr_info[-1]
                res.base_currency = curr_info[0]

            if len(parts) == 4:
                res.kind = ContratType.Option
                res.strike = float(str(parts[2]).replace('d', '.'))
                res.option_type = OptionType.Call if parts[3] == "C" else OptionType.Put
                datestr = parts[1]

            elif len(parts) == 2 or inst_name.endswith('PERPETUAL'):
                res.kind = ContratType.Future
                datestr = None if inst_name.endswith('PERPETUAL') else parts[1]

            elif len(parts) == 1 and len(curr_info) == 2:
                res.kind = ContratType.Spot
            else:
                return None

            if datestr:
                year = datestr[-2:]
                month = months.index(datestr[-5:-2]) + 1
                day = int(datestr[:-5])
                time_str = "20%s-%02d-%02d 08:00:00" % (year, month, day)
                res.expire_timestamp = datetime.strptime(
                    time_str, "%Y-%m-%d %H:%M:%S").replace(tzinfo=pytz.utc).timestamp() * 1000
            return res
        except Exception as exc:
            logging.exception(f'parse inst name {inst_name} except: {exc}')

    @classmethod
    def transfer_summary_data(cls, summary_data: Dict, index_price: Optional[float] = None) -> CurrencySummary:
        currency = summary_data['currency']
        currency_summary = CurrencySummary(margin_currency=summary_data['currency'], )

        if currency not in STABLE_COINS:
            currency_summary.base_currency = currency

        currency_summary.equity = summary_data['equity']
        is_pm = summary_data.get('portfolio_margining_enabled', False)
        is_cross = summary_data.get('cross_collateral_enabled', False)
        currency_summary.im = summary_data.get('initial_margin', 0)
        currency_summary.mm = summary_data.get('projected_maintenance_margin', 0)
        if is_cross:
            balance = summary_data.get('total_equity_usd') if is_pm else summary_data.get('total_margin_balance_usd')
            currency_summary.im_ratio = summary_data.get('total_initial_margin_usd', 0) / balance if balance else 0
            currency_summary.mm_ratio = summary_data.get('total_maintenance_margin_usd', 0) / balance if balance else 0
            currency_summary.equity_usd = summary_data['total_equity_usd']
        else:
            balance = summary_data.get('equity') if is_pm else summary_data.get('balance')
            currency_summary.equity_usd = currency_summary.equity * index_price if index_price else None
            currency_summary.im_ratio = currency_summary.im / balance if balance else 0
            currency_summary.mm_ratio = currency_summary.mm / balance if balance else 0

        currency_summary.is_cross = is_cross
        currency_summary.is_pm = is_pm

        currency_summary.delta = summary_data.get('delta_total')
        currency_summary.gamma = summary_data.get('options_gamma')
        currency_summary.vega = summary_data.get('options_vega')
        currency_summary.theta = summary_data.get('options_theta')
        return currency_summary

    @classmethod
    def transfer_position_data(cls, position_data: Dict) -> Position:
        return Position(
            instrument=position_data['instrument_name'],
            size=abs(position_data['size']),
            direction=position_data['direction'],
            avg_price=position_data['average_price'],
            im=str2float(position_data.get('initial_margin')),
            mm=str2float(position_data.get('maintenance_margin')),
            mark_price=str2float(position_data.get('mark_price')),

            delta=str2float(position_data.get('delta')),
            gamma=str2float(position_data.get('gamma')),
            vega=str2float(position_data.get('vega')),
            theta=str2float(position_data.get('theta')),
        )

    @classmethod
    def order_state_transfer(cls, order_state):
        if order_state == 'open':
            return OrderState.Open
        if order_state == 'filled':
            return OrderState.Filled
        if order_state == 'cancelled':
            return OrderState.Cancelled
        if order_state == 'rejected':
            return OrderState.Cancelled
        if order_state == 'untriggered':
            return OrderState.Open
        return order_state

    @classmethod
    def transfer_order_data(cls, order_data) -> Order:
        return Order(
            order_id=order_data['order_id'], instrument=order_data['instrument_name'],
            direction=order_data['direction'], label=order_data['label'],
            amount=order_data['amount'], filled_amount=order_data['filled_amount'],
            price=order_data['price'], order_type=order_data['order_type'],
            state=cls.order_state_transfer(order_data['order_state']),
            avg_price=order_data['average_price'],
            create_timestamp=order_data['creation_timestamp'],
            post_only=order_data['post_only'],
            reduce_only=order_data['reduce_only'],
            update_timestamp=order_data['last_update_timestamp'],
        )

    @classmethod
    def transfer_trade_data(cls, trade_data):
        return Trade(
            instrument=trade_data['instrument_name'],
            order_id=trade_data['order_id'],
            trade_id=trade_data['trade_id'],
            price=trade_data['price'],
            timestamp=trade_data['timestamp'],
            fee_currency=trade_data['fee_currency'],
            fee=trade_data['fee'],
            direction=trade_data['direction'],
            order_type=trade_data['order_type'],
            mark_price=trade_data['mark_price']
        )


def _handle_api_exception(func):
    """装饰器：统一处理API异常"""
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except rest.ApiException as e:
            if e.status == 0:  # 连接失败、超时等网络错误
                raise ExchangeErrorCode.ServerUnavailable.to_error()
            # 其他 API 错误按原有逻辑处理
            raise DeribitTools.transfer_error(e)
        except Exception as e:
            logging.exception(f'Deribit API call failed: {str(e)}')
            raise
    return wrapper


class DeribitPublicService(ExchangePublicServiceBase):

    platform = Exchange.DERIBIT

    def __init__(self, *args, **kwargs):
        super(DeribitPublicService, self).__init__(*args, **kwargs)
        exchange_config = EXCHANGE_CONFIG.get(self.env)
        if not exchange_config:
            raise ValueError(f'{self.platform} {self.env} has not exchange config.')
        http_node = exchange_config['api_host']
        conf = configuration.Configuration(host=http_node)
        self._public_api_client = public_api.PublicApi(api_client.ApiClient(conf))
        self._quote_catcher = None
        self._cache = cacheout.Cache()

    @property
    def name(self):
        return 'epubs:dbt'

    def init_quote_catcher(self, base_currencies, margin_currencies):
        if not self._quote_catcher:
            self._quote_catcher = DbtQuotesCatcher(
                env=self.env, logger=self._logger, currencies=base_currencies, margin_currencies=margin_currencies,)
            self._quote_catcher.run()
            self._logger.info(f'{self.platform} quote catcher is inited.')
        else:
            self._logger.warning(f'{self.platform} quote catcher is running, can not repeat init it.')

    @_handle_api_exception
    def index_price(self, pair: IndexPair) -> IndexPrice:
        index_name = f'{pair.base}_{pair.quote}'.lower()
        res = self._quote_catcher.get_index_price(index_name)
        if res:
            return res

        res = self._public_api_client.public_get_index_price_get(index_name=index_name).get("result")
        return IndexPrice(price=res['index_price'], timestamp=int(time.time() * 1000))

    @_handle_api_exception
    def ticker(self, instrument: str, **kwargs) -> Ticker:
        ticker_data = self._public_api_client.public_ticker_get(instrument_name=instrument).get("result")
        return DeribitTools.transfer_ticker_data(ticker_data)

    @_handle_api_exception
    def mark_price(self, instrument, **kwargs) -> float:
        mark_price = self._quote_catcher.mark_price(instrument)
        if mark_price is not None:
            return mark_price

        ticker_data = self._public_api_client.public_ticker_get(instrument_name=instrument).get("result")
        return ticker_data['mark_price']

    @_handle_api_exception
    def orderbook(self, instrument, **kwargs) -> OrderBook:
        book = self._public_api_client.public_get_order_book_get(instrument_name=instrument).get("result")
        return OrderBook(
            asks=[OrderBookItem(*item) for item in book.get('asks') or []],
            bids=[OrderBookItem(*item) for item in book.get('bids') or []],
            timestamp=book.get('timestamp'),
        )

    def get_instruments_from_catcher(self):
        instruments = self._quote_catcher.get_instruments()
        if not instruments:
            return
        return instruments

    @_handle_api_exception
    def instruments_info(self, base_currency=None, margin_currency=None, kind=None, use_cache=True, **kwargs) -> List[Instrument]:
        results = []
        if use_cache:
            results = self.get_instruments_from_catcher()
            if results is not None:
                return results

        cache_key = f'{self.platform}:instrumentsInfo'
        if not results:
            results = self._cache.get(cache_key, ) if use_cache else None
        if not results:
            try:
                resp = self._public_api_client.public_get_instruments_get('any')
                instruments = resp.get('result')
            except Exception as e:
                self._logger.exception(f'query instruments for {margin_currency} except: {e}')
                instruments = []

            results = []
            for instrument_data in instruments:
                instrument = DeribitTools.transfer_instrument_data(instrument_data)
                if instrument:
                    results.append(instrument)
            if results:
                self._cache.set(cache_key, results, ttl=3600)

        _results = []
        for result in results:
            if base_currency is not None and result.base_currency != base_currency:
                continue
            if margin_currency is not None and result.margin_currency != margin_currency:
                continue
            if kind is not None and result.kind != kind:
                continue
            _results.append(result)
        return _results

    def instrument_info(self, inst, **kwargs) -> Instrument:
        inst_info = self._quote_catcher.get_instrument(inst)
        if inst_info:
            return inst_info

        inst_simple_info = self.parse_inst(inst)
        if inst_simple_info:
            currency = inst_simple_info.base_currency
            instruments = self.instruments_info(
                base_currency=currency, margin_currency=inst_simple_info.margin_currency, kind=inst_simple_info.kind)
            for inst_info in instruments:
                if inst_info.instrument_name == inst:
                    return inst_info

    @classmethod
    def parse_inst(cls, inst_name, **kwargs) -> Optional[SimpleInstrument]:
        return DeribitTools.parse_inst(inst_name)

    def get_futures_info(self):
        return self._quote_catcher.get_futures_info()

    def get_future_found_rate_details(self):
        return self._quote_catcher.get_future_found_rate_details()


@dataclass
class TokenInfo:
    refresh_token: str
    scope: str
    access_token: str
    expire_ts: int

    def is_expired(self):
        return time.time() > self.expire_ts


class DBTUserDataWsCatcher(DeribitWSRunner, ExchangeUserDataCatcher, ):

    platform = Exchange.DERIBIT
    MAX_AUTH_FAIL_TIMES = 5

    def __init__(self, auth_info: AuthInfo, env, logger=None, *args, **kwargs):

        if not isinstance(auth_info, ApiAuthInfo):
            raise ValueError(f'{self.platform} user data catcher only support api account.')
        self._auth_info = auth_info
        if env not in EXCHANGE_CONFIG:
            raise ValueError(f'{self.platform} got invalid args for env: {env}')
        ws_url = EXCHANGE_CONFIG[env]['ws_host']
        self._account_summaries: Dict[str, CurrencySummary] = {}
        self._positions: Dict[str, Position] = {}
        self._open_orders: Dict[str, Order] = {}
        self._token_info: Optional[TokenInfo] = None

        self._auth_failed_times = 0
        self._scheduler: Optional[asyncio.Task] = None

        DeribitWSRunner.__init__(
            self, env=env, ws_url=ws_url, logger=logger,
            *args, **kwargs
        )
        ExchangeUserDataCatcher.__init__(self, *args, **kwargs)

    @property
    def name(self):
        return f'eudc:dbt:{self._auth_info.key()[-8:]}'

    async def _on_connected(self):
        self._auth_failed_times = 0
        await self._auth()

    async def _auth(self):
        if self.is_auth:
            await self._request('public/auth', params={
                "grant_type": "refresh_token",
                "refresh_token": self._token_info.refresh_token,
            }, callback=self._on_auth)
        else:
            await self._request('public/auth', params={
                "grant_type": "client_credentials",
                "client_id": self._auth_info.api_key,
                "client_secret": self._auth_info.api_secret,
            }, callback=self._on_auth)

    @property
    def is_auth(self):
        return self._token_info and not self._token_info.is_expired()

    def _should_be_run(self, connected=False):
        return self._auth_failed_times < self.MAX_AUTH_FAIL_TIMES or self.is_auth

    async def _on_auth(self, res):
        if isinstance(res, ApiException):
            self._auth_failed_times += 1
            if self._auth_failed_times < self.MAX_AUTH_FAIL_TIMES:
                self._loop.call_later(5, asyncio.create_task, self._auth())
            return

        self._auth_failed_times = 0
        self._token_info = TokenInfo(
            access_token=res['access_token'],
            scope=res['scope'],
            refresh_token=res['refresh_token'],
            expire_ts=int(time.time()) + res['expires_in'] - 10,
        )
        self._logger.info(f'{self.name} auth success.')
        self._scheduler = self._loop.call_later(
            int(res['expires_in'] / 3), self._loop.create_task, self._refresh_auth_info())

        await self._request('/private/get_account_summaries', params={}, callback=self._handle_req_summaries)
        await self._request('/private/get_open_orders', params={}, callback=self._handle_orders_req)
        await self._request('/private/get_positions', params={}, callback=self._handle_req_positions)

        await self._subscribe(["user.portfolio.any", "user.changes.any.any.100ms"], is_pub=False)

    async def _refresh_auth_info(self):
        self._logger.info(f'refresh auth info for {self.name}')
        while self.is_available():
            await self._auth()

    async def _on_disconnect(self):
        if self._scheduler and not self._scheduler.done():
            self._scheduler.cancel()

        self._account_summaries = None
        self._positions = None
        self._open_orders = None

    async def _handle_subscription_msg(self, channel, data):
        if channel.startswith('user.portfolio.'):
            await self._on_summaries_update(data)
        elif channel.startswith('user.changes.'):
            orders = data.get('orders')
            if orders:
                await self._on_orders_update(orders)

            positions = data.get('positions')
            if positions:
                await self._on_positions_update(positions)

    async def _handle_req_summaries(self, res, ):
        if isinstance(res, ApiException):
            self._logger.error(f'request positions except: {res}')
        self._account_summaries = {}
        summaries = res.get('summaries')
        await self._on_summaries_update(summaries, )

    async def _on_summaries_update(self, res, ):
        if self._account_summaries is None:
            return

        summaries = [res] if isinstance(res, dict) else res

        for summary in summaries:
            currency_summary = DeribitTools.transfer_summary_data(summary)
            self._account_summaries[currency_summary.margin_currency] = currency_summary

    async def _handle_req_positions(self, res):
        if isinstance(res, ApiException):
            self._logger.error(f'request positions except: {res}')
        self._positions = {}
        await self._on_positions_update(res)

    async def _on_positions_update(self, positions, ):
        if self._positions is None:
            return

        for position_data in positions:
            pos = DeribitTools.transfer_position_data(position_data)
            if pos.size <= 0:
                self._positions.pop(pos.instrument, None)
            else:
                self._positions[pos.instrument] = pos

    async def _handle_orders_req(self, res, ):
        if isinstance(res, ApiException):
            self._logger.error(f'request open orders except: {res}')
        self._open_orders = {}
        await self._on_orders_update(res, )

    async def _on_orders_update(self, data, ):
        if self._open_orders is None:
            return
        data = data if isinstance(data, list) else [data]

        for order_data in data:
            order = DeribitTools.transfer_order_data(order_data)
            if order.state in {OrderState.Filled, OrderState.Cancelled}:
                self._open_orders.pop(order.order_id, None)
            elif order.state == OrderState.Open:
                self._open_orders[order.order_id] = order
            else:
                self._logger.warning(f'{self.name} get unknown order state {order_data}')

    def account_summary(self, margin_currency, **kwargs) -> CurrencySummary:
        return self._account_summaries.get(margin_currency)

    def account_summaries(self, margin_currencies: Optional[List] = None, **kwargs):
        if self.is_available() and self._account_summaries is not None:
            return list(self._account_summaries.values())

    def positions(self, margin_currency=None, **kwargs):
        if self.is_available():
            return list(self._positions.values())

    def open_orders(self, margin_currency=None, **kwargs):
        if self.is_available():
            return list(self._open_orders.values())

    def is_available(self, ):
        return super().is_available() and self.is_auth


class DeribitPriService(ExchangePriServiceBase):
    platform = Exchange.DERIBIT
    __broker_info: Dict = None

    @classmethod
    def install_broker_info(cls, client_id, client_secret, ):
        if cls.__broker_info:
            raise ValueError(f'broker info is installed, can not repeat install.')

        if not client_id and not client_secret:
            raise ValueError(f'broker info must with client_id and client_secret')
        cls.__broker_info = {
            'client_id': client_id, 'client_secret': client_secret,
        }

    def __init__(self, auth_info: AuthInfo, env='test', *args, **kwargs):
        if auth_info.type not in [AuthInfoType.Api, AuthInfoType.Oauth]:
            raise ValueError(f'{self.platform} private service only support `api` and `oauth` type auth info.')
        super(DeribitPriService, self).__init__(auth_info, env, *args, **kwargs)

        exchange_config = EXCHANGE_CONFIG.get(self.env)
        self._api_host = exchange_config['api_host']
        conf = configuration.Configuration(host=self._api_host)
        self._private_api = PrivateApi(api_client.ApiClient(conf))
        self._token_info: Optional[TokenInfo] = None
        self._auth_lock = threading.Lock()
        self._user_data_catcher = DBTUserDataWsCatcher(self._auth_info, env=self.env, logger=self._logger)

    @property
    def name(self):
        return f'epris:dbt:{self._auth_info.key()[-8:]}'

    def start_catcher_thread(self):
        self.ensure_catcher_run()

    def ensure_catcher_run(self):
        if not self._user_data_catcher.is_run():
            self._user_data_catcher.run()

    def is_catcher_available(self):
        return self._user_data_catcher.is_available()

    def start_catcher_asyncio(self, loop: asyncio.BaseEventLoop):
        return asyncio.run_coroutine_threadsafe(self._user_data_catcher.run_as_coroutine(), loop=loop)

    @property
    def private_api(self):
        with self._auth_lock:
            now = time.time()
            if not self._token_info or self._token_info.expire_ts < now:
                self._update_token_info()
            if self._private_api.api_client.configuration.access_token != self._token_info.access_token:
                self._private_api.api_client.configuration.access_token = self._token_info.access_token
            return self._private_api

    def check_auth(self):
        self._update_token_info()

    def _update_token_info(self):
        if not self.__broker_info:
            raise ValueError(f'{self.platform} broker info is required in current version, please install first.')

        timestamp = int(time.time()) * 1000
        payloads = {'method': 'public/auth', 'params': {'grant_type': 'app_user'}}
        if isinstance(self._auth_info, DeribitOauthAuthInfo):
            payloads['params']['user_id'] = self._auth_info.oauth_user_id
        elif isinstance(self._auth_info, ApiAuthInfo):

            timestamp = int(time.time()) * 1000
            nonce = ''
            plain_text = ("%d\n%s\n%s" % (timestamp, nonce, ''))
            signature = hmac.new(
                bytes(self._auth_info.api_secret, encoding='utf8'),
                bytes(plain_text, encoding='utf-8'),
                digestmod=hashlib.sha256
            ).hexdigest()

            payloads['params']['client_id'] = self._auth_info.api_key
            payloads['params']['timestamp'] = timestamp
            payloads['params']['signature'] = signature
            payloads['params']['nonce'] = nonce
            payloads["params"]["data"] = ''
        else:
            raise ValueError(f'{self.platform} private service only support `api` and `oauth` type auth info.')

        broker_client_id = self.__broker_info['client_id']
        broker_client_secret = self.__broker_info['client_secret']
        broker_nonce = ''
        broker_data = json.dumps(payloads).replace(' ', '')
        broker_plain_text = f"{timestamp}\n{broker_nonce}\nPOST\n/api/v2\n{broker_data}\n"
        broker_signature = hmac.new(
            bytes(broker_client_secret, encoding='utf8'),
            bytes(broker_plain_text, encoding='utf-8'),
            digestmod=hashlib.sha256
        ).hexdigest()
        headers = {
            'authorization': f"app-deri-hmac-sha256 id={broker_client_id},"
                             f"ts={timestamp},nonce={broker_nonce},sig={broker_signature}"
        }
        try:
            resp = requests.post(self._api_host, data=broker_data, headers=headers, timeout=5)
            resp = resp.json()
            if resp.get('error'):
                error = resp.get('error')
                if error.get('code') in DeribitTools.EXCHANGE_CODE_TO_GREEKS_CODE_MAP:
                    raise DeribitTools.EXCHANGE_CODE_TO_GREEKS_CODE_MAP[error.get('code')].to_error()
                raise DeribitTools.EXCHANGE_CODE_TO_GREEKS_CODE_MAP['default'].to_error()

            result = resp.get('result')
            self._token_info = TokenInfo(
                access_token=result['access_token'],
                scope=result['scope'],
                refresh_token=result['refresh_token'],
                expire_ts=int(time.time()) + result['expires_in'] - 10,
            )
        except requests.Timeout:
            raise ExchangeErrorCode.ServerRequestTimeout.to_error()
        except Exception as e:
            self._logger.exception(f'{self.platform} refresh auth token failed: {e}')
            raise e

    @_handle_api_exception
    def account_summaries(self, **kwargs) -> List[CurrencySummary]:
        if self._user_data_catcher.is_available():
            summaries = self._user_data_catcher.account_summaries()
            if summaries:
                self._logger.info(f'{self.name} get summaries from catcher')
                return summaries

        res = self.private_api.private_get_account_summaries_get()
        results = []
        summaries = res['result']['summaries']

        for summary in summaries:
            currency_summary = DeribitTools.transfer_summary_data(summary)
            results.append(currency_summary)
        return results

    @_handle_api_exception
    def account_summary(self, base_currency=None, margin_currency=None, **kwargs) -> CurrencySummary:
        if not margin_currency:
            raise ValueError('margin currency is required.')

        res = self.private_api.private_get_account_summary_get(margin_currency)
        return DeribitTools.transfer_summary_data(res['result'])

    def get_positions_from_catcher(self, margin_currency=None):
        if self._user_data_catcher.is_available():
            positions = self._user_data_catcher.positions(margin_currency)
            return positions

    @_handle_api_exception
    def positions(self, margin_currency=None, **kwargs) -> List[Position]:
        positions = self.get_positions_from_catcher(margin_currency=margin_currency)
        if positions is not None:
            self._logger.info(f'{self.name} get positions from catcher')
            return positions

        data = self.private_api.private_get_positions_get(margin_currency)['result']
        results = []
        for position_data in data:
            results.append(DeribitTools.transfer_position_data(position_data))
        return results

    def get_open_orders_from_catcher(self, margin_currency=None, **kwargs):
        if self._user_data_catcher.is_available():
            open_orders = self._user_data_catcher.open_orders(margin_currency)
            return open_orders

    @_handle_api_exception
    def open_orders(self, margin_currency=None, **kwargs) -> List[Order]:
        open_orders = self.get_open_orders_from_catcher(margin_currency)
        if open_orders is not None:
            self._logger.info(f'{self.name} get open orders from catcher')
            return open_orders

        if margin_currency:
            data = self.private_api.private_get_open_orders_by_currency_get(margin_currency)['result']
        else:
            data = self.private_api.private_get_open_orders_get()['result']
        results = []
        for order_data in data:
            results.append(DeribitTools.transfer_order_data(order_data))
        return results

    @_handle_api_exception
    def create_order(self, instrument, size, price=None, order_type=OrderType.Limit, price_type=None, reduce_only=False,
                     post_only=False, label=None, **kwargs):
        # ToDo 暂时只考虑期货下单参数
        if not instrument or not size:
            raise ValueError('create order but params is invalid.')

        kwargs = {
            'instrument_name': instrument, 'amount': abs(size), 'price': price, 'type': order_type,
            'time_in_force': 'good_til_cancelled',
            'reduce_only': 'true' if reduce_only else 'false',
            'post_only': 'true' if post_only else 'false', 'label': label
        }
        kwargs = {key: value for key, value in kwargs.items() if value is not None}

        if size > 0:
            res = self.private_api.private_buy_get(**kwargs)
        else:
            res = self.private_api.private_sell_get(**kwargs)
        return DeribitTools.transfer_order_data(res['result']['order'])

    @_handle_api_exception
    def reset_order(self, order_id, size=None, price=None, **kwargs):
        if size is None and not price:
            raise ValueError('size or price is required.')
        resp = self.private_api.private_edit_get(order_id=order_id, amount=size, price=price)
        return DeribitTools.transfer_order_data(resp['result']['order'])

    @_handle_api_exception
    def cancel_order(self, order_id=None, label=None, **kwargs):
        if not order_id and not label:
            raise ValueError('order id or label is required when cancel order')
        if order_id:
            res = self.private_api.private_cancel_get(order_id)
        else:
            res = self.private_api.private_cancel_by_label_get(label=label)
        return DeribitTools.transfer_order_data(res.get('result'))

    @_handle_api_exception
    def order_trades(self, order_id) -> List[Trade]:
        results = self._private_api.private_get_user_trades_by_order_get(order_id)['result']
        return [result for result in results]

    @_handle_api_exception
    def order_state(self, order_id, **kwargs):
        if not order_id:
            raise ValueError(f'order_id is required when get {self.platform} order state')

        result = self.private_api.private_get_order_state_get(order_id).get('result')
        return DeribitTools.transfer_order_data(result)

